package com.heisen.rocketmq;

import com.alibaba.fastjson.JSON;
import com.heisen.rocketmq.model.BaseMessage;
import com.heisen.rocketmq.utils.RocketMQUtils;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.client.producer.SendStatus;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;

/**
 * @author sen
 * @description  rocket mq 增强监听类
 * @date 2023/8/8 15:43
 */
@Slf4j
public abstract class EnhanceMessageHandler implements RocketMQListener<BaseMessage> {

    /**
     * 默认重试次数
     */
    private static final int MAX_RETRY_NUM = 3;

    /**
     * 重试时间  每60秒重试
     */
    private static final int RETRY_TIME = 60;

    /**
     * 消息处理
     *
     * @param message 待处理消息
     * @throws Exception 消费异常
     */
    protected abstract void handleMessage(String message) throws Exception;


    /**
     * 超过重试次数消息，需要启用isRetry
     * 具体漏记重新这个方法
     *
     * @param message 待处理消息
     */
    protected void handleMaxRetriesExceeded(BaseMessage message) {
        log.info("【超过最大重试次数  结束重试】 id:{}", message.getId());
    }

    ;


    /**
     * 是否需要根据业务规则过滤消息，去重逻辑可以在此处处理
     *
     * @param message 待处理消息
     * @return true: 本次消息被过滤，false：不过滤
     */
    protected boolean filter(BaseMessage message) {
        return false;
    }

    /**
     * 是否异常时重复发送
     *
     * @return true: 消息重试，false：不重试
     */
    protected boolean isRetry() {
        return false;
    };

    /**
     *  告警事件  isAlarm 为true是会触发
     * @param message 待处理告警事件消息
     */
    protected  void handleAlarmMessage(String message){
        log.info("【告警事件】");
    };

    /**
     * 是否告警
     * @return true 会触发handleAlarmMessage方法
     */
    protected  boolean isAlarm(){
        return false;
    };

    /**
     * 消费异常时是否抛出异常
     * 返回true，则由rocketmq机制自动重试
     * false：消费异常(如果没有开启重试则消息会被自动ack) 如果isRetry是true  程序会自动重试
     */
    protected boolean throwException() {
        return false;
    }

    ;


    /**
     * 最大重试次数
     *
     * @return 最大重试次数，默认3次
     */
    protected int getMaxRetryNum() {
        return MAX_RETRY_NUM;
    }

    protected int getRetryTime() {
        return RETRY_TIME;
    }


    /**
     * 消费夏夏
     *
     * @param message
     */
    @Override
    public void onMessage(BaseMessage message) {
        dispatchMessage(message);
    }


    /**
     * 使用模板模式构建消息消费框架，可自由扩展或删减
     */
    public void dispatchMessage(BaseMessage message) {
        // 基础日志记录被父类处理了
        log.info("【消费者收到消息】 message{}", JSON.toJSONString(message));

        if (filter(message)) {
            log.info("消息id{}不满足消费条件，已过滤。", message.getId());
            return;
        }
        // 超过最大重试次数时调用子类方法处理
        if (message.getNum() >= getMaxRetryNum()) {
            handleMaxRetriesExceeded(message);
            return;
        }
        try {
            long now = System.currentTimeMillis();

            if(message.getData() instanceof  String ){
                handleMessage(message.getData().toString());
            }else{
                handleMessage(JSON.toJSONString(message.getData()));
            }

            long costTime = System.currentTimeMillis() - now;
            log.info("【消息{}消费成功，耗时[{}ms]】", message.getId(), costTime);
        } catch (Exception e) {
            log.error("【消息{}消费异常】 e:{}", message.getId(), e);
            // 是捕获异常还是抛出，由子类决定
            if (throwException()) {
                //抛出异常，由DefaultMessageListenerConcurrently类处理
                throw new RuntimeException(e);
            }
            //此时如果不开启重试机制，则默认ACK了
            if (isRetry()) {
                handleRetry(message);
            }
        }finally {
            // 告警事件
            if(isAlarm()){
                try {

                    if(message.getData() instanceof  String ){
                        handleAlarmMessage(message.getData().toString());
                    }else{
                        handleAlarmMessage(JSON.toJSONString(message.getData()));
                    }
                }catch (Exception e){
                    log.info("【告警异常】 e:{}", e);
                }

            }
        }
    }

    protected void handleRetry(BaseMessage message) {
        // 获取子类RocketMQMessageListener注解拿到topic和tag
        RocketMQMessageListener annotation = this.getClass().getAnnotation(RocketMQMessageListener.class);
        if (annotation == null) {
            return;
        }
        //重新构建消息体
        message.setNum(message.getNum() + 1);
        SendResult sendResult;
        try {
            //    // 如果消息发送不成功，则再次重新发送，如果发送异常则抛出由MQ再次处理(异常时不走延迟消息)
            if (null != annotation.selectorExpression()) {
                sendResult = RocketMQUtils.sendDelayed(annotation.topic(), annotation.selectorExpression(), message, getRetryTime());
            } else {
                sendResult = RocketMQUtils.sendDelayed(annotation.topic(), message, getRetryTime());
            }
            log.info("【消息 [{}] 失败重试中】 num:{}", message.getId(), message.getNum());
        } catch (Exception ex) {
            // 此处捕获之后，相当于此条消息被消息完成然后重新发送新的消息
            //由生产者直接发送
            throw new RuntimeException(ex);
        }
        // 发送失败的处理就是不进行ACK，由RocketMQ重试
        if (sendResult.getSendStatus() != SendStatus.SEND_OK) {
            throw new RuntimeException("重试消息发送失败");
        }

    }

}
